#include"CCClientProcess.h"

CCClientProcess::CCClientProcess(){
    m_clientsockfd = -1;
    m_bClientRunning = true;
    pthread_mutex_init(&m_recvMutex,NULL);
    pthread_mutex_init(&m_sendMutex,NULL);
}
CCClientProcess::~CCClientProcess(){
    std::cout<<"~CCClientProcess used sockedfd:"<<m_clientsockfd<<std::endl;
    closeListenSocketfd();
    freeAllRemainAVStream();
    
}
void CCClientProcess::StartClientProcessWithSockfd(int fd){
    m_clientsockfd = fd;
    
    detach_thread_creat(NULL,(void*)startRecvThread,(void*)this);
    detach_thread_creat(NULL,(void*)startRSendThread,(void*)this);

}

void CCClientProcess::startRecvThread(long long num){
    CCClientProcess* clientptr = (CCClientProcess*)num;
    if(clientptr)
        clientptr->RunRecvThread();
}
void CCClientProcess::startRSendThread(long long num){
    CCClientProcess* clientptr = (CCClientProcess*)num;
    if(clientptr)
    {
        //printf("startRSendThread-------------------------------\n");
        clientptr->RunSendThread();
    }
        
}
void CCClientProcess::RunRecvThread(){
    int keepAliveCount = 5;
    while(m_bClientRunning){
        usleep(2000*1000);
        keepAliveCount--;
        if(keepAliveCount<0){
            LoseAlive();
            break;
        }
        CC_MsgHeader msgheader;
        memset((char*)&msgheader,0,sizeof(CC_MsgHeader));
        recvSocketData((char*)&msgheader,sizeof(CC_MsgHeader));
        if(strcmp((char*)&msgheader.headrID,"GUGU")==0){
            if(msgheader.msgType==MSGHEADER_TYPE_KEEPALIVE){
                printf("keeping alive~~\n");
                keepAliveCount = 5;
            }
            
        }
    }
}
void CCClientProcess::RunSendThread(){
    while (m_bClientRunning)
    {
        if(m_avStreamQueue.size()==0){
            
            usleep(10*1000);
            continue;
        }
        CC_AVStream* pAVStream = (CC_AVStream*)m_avStreamQueue.dequeue();
        if((pAVStream!=NULL)&&(pAVStream->buffer!=NULL)&&(pAVStream->size>0)){
            
            CC_MsgHeader header;
            memset((char*)&header,0,sizeof(CC_MsgHeader));
            strncpy((char*)&header.headrID,"GUGU",sizeof(header.headrID));
            //printf("header.headrID:%s\n",(char*)header.headrID);
            header.msgType = MSGHEADER_TYPE_AVSTREAM;
            header.subType = CONTEXT_AVSTREAM_VIDEO;
            header.contentLength = pAVStream->size;
            //printf("header: ID:%s,msgType:%d,header.subType:%d,contentLen:%d\n",header.headrID,header.msgType,header.subType,header.contentLength);
            sendSocketData((char*)&header,sizeof(CC_MsgHeader));
            sendSocketData((char*)&pAVStream->buffer,pAVStream->size);

            free(pAVStream->buffer);
            delete pAVStream;

        }


    }
    
}

bool CCClientProcess::recvSocketData(char* pbuff,unsigned int length){
    signal(SIGPIPE,SIG_IGN);
    pthread_mutex_lock(&m_recvMutex);
    int recv_len = 0;
    int nRet = 0;

    while(recv_len<length){
        nRet = recv(m_clientsockfd,pbuff,length-recv_len,0);

        if(nRet<0){
            if(errno == EAGAIN||errno == EINTR||errno==EWOULDBLOCK){
                usleep(10*1000);
                continue;
            }
        }

        if(nRet==-1||nRet==0){
            pthread_mutex_unlock(&m_recvMutex);
            printf("cSeocket recv error");
            LoseAlive();
            return false;
        }
        recv_len+=nRet;
        pbuff+=nRet;
    }
    pthread_mutex_unlock(&m_recvMutex);
    return true;
}
bool CCClientProcess::sendSocketData(char* pbuff,unsigned int length){
    signal(SIGPIPE,SIG_IGN);
    pthread_mutex_lock(&m_sendMutex);
    int send_len = 0;
    int nRet = 0;

    while(send_len<length){
        if(m_clientsockfd>0){
            nRet = send(m_clientsockfd,pbuff,length-send_len,0);
            if(nRet<0){
                if(errno == EAGAIN||errno == EINTR||errno==EWOULDBLOCK){
                    usleep(10*1000);
                    continue;
                }
                
                pthread_mutex_unlock(&m_recvMutex);
                printf("cSeocket recv error");
                return false;
            }
            if(nRet==0){
                pthread_mutex_unlock(&m_recvMutex);
                LoseAlive();
                return false;
            }

            send_len+=nRet;
            pbuff+=nRet;

        }else{
            printf("socket fd error %d\n",m_clientsockfd);
            pthread_mutex_unlock(&m_sendMutex);
            LoseAlive();
            return false;
        }

        
    }
    pthread_mutex_unlock(&m_sendMutex);
    return true;
}

void CCClientProcess::LoseAlive(){
    m_bClientRunning = false;
    printf("Client Timeout Lose Alive\n");
}

bool CCClientProcess::getClientAliveStatus(){
    return m_bClientRunning;
}

void CCClientProcess::EnqueueAVStream(uint8_t* buff,uint32_t buffsize,uint16_t type){
    if(m_avStreamQueue.size()>50){
        return;
    }

    CC_AVStream* pAVStream = new CC_AVStream();
    pAVStream->buffer = (uint8_t*)malloc(buffsize);
    memcpy(pAVStream->buffer,buff,buffsize);
    pAVStream->size = buffsize;
    pAVStream->type = type;

    m_avStreamQueue.enqueue(pAVStream);
}

void CCClientProcess::freeAllRemainAVStream(){
    while (!m_avStreamQueue.isEmpty())
    {
        CC_AVStream* pAVStream = (CC_AVStream*)m_avStreamQueue.dequeue();
        if((pAVStream!=NULL)&&(pAVStream->buffer!=NULL)){
            free(pAVStream->buffer);
            delete pAVStream;

        }
    }
    
}

void CCClientProcess::closeListenSocketfd(){
    if(m_clientsockfd){
        close(m_clientsockfd);
    }
}